package com.etc

import org.apache.spark.{SparkConf, SparkContext}


object Action {
  //将数据聚合成一个元素
  def reduce(): Unit ={
    val list = List(1,2,3,4,5,6,7,8,9,10)
    val i = list.reduce(_ + _)
    println(i)

  }
  //将RDD的数据收集到driver端。数据量大的数据会报OOM。建议不要这样做，通常用take(num:Int)来代替
  def collect(): Unit ={
    val cont = new SparkConf().setAppName("collect").setMaster("local")
    val sc = new SparkContext(cont)
    val list = List(1,2,3,4,5,6,7,8,9,10)
    val ints = sc.parallelize(list).map(a => a * 2).collect()
    for(i <- ints){
      println(i)
    }
    sc.stop()
  }
  //返回rdd的元素个数
  def count(): Unit ={
    val cont = new SparkConf().setAppName("count").setMaster("local")
    val sc = new SparkContext(cont)
    val list = List(1,2,3,4,5,6,7,8,9,10)
    val l = sc.parallelize(list).count()
    println(l)
    sc.stop()
  }
  //设置输出的个数
  def take(): Unit ={
    val cont = new SparkConf().setAppName("take").setMaster("local")
    val sc = new SparkContext(cont)
    val list = List(1,2,3,4,5,6,7,8,9,10)
    sc.parallelize(list).take(6).foreach(println)
    sc.stop()
  }
  /*函数将数据输出，存储到 HDFS 的指定目录。
  将 RDD 中的每个元素映射转变为 (null， x.toString)，
  然后再将其写入 HDFS。图 中左侧方框代表 RDD 分区，
  右侧方框代表 HDFS 的 Block。通过函数将RDD 的每个分区存储为 HDFS 中的一个 Block
*/
  def saveAsTextFile(): Unit ={
    val cont = new SparkConf().setAppName("saveAsTextFile").setMaster("local")
    val sc = new SparkContext(cont)
    val list = List(1,2,3,4,5,6,7,8,9,10)
    sc.parallelize(list).map(a => a * 2).saveAsTextFile("jiege")
    sc.stop()
  }
  //根据key来累计value的个数
  def countByKey(): Unit ={
    val cont = new SparkConf().setAppName("countByKey").setMaster("local")
    val sc = new SparkContext(cont)
    val list = List(
      new Tuple2("class1", "leo"),
      new Tuple2("class2", "jack"),
      new Tuple2("class1", "marry"),
      new Tuple2("class2", "tom"),
      new Tuple2("class2", "david")
    )
    sc.parallelize(list).countByKey().foreach(println)
    sc.stop()
  }
    //元祖去重
    def collectAsMap(): Unit ={
      val cont = new SparkConf().setAppName("collectAsMap").setMaster("local")
      val sc = new SparkContext(cont)
      sc.parallelize(Array((1,12),(2,11),(1,12),(2,11),(3,13))).collectAsMap().foreach(println)
      sc.stop()
    }
    //根据key搜索valuelookup（key：K）：Seq[V]
  //Lookup函数对（Key，Value）型的RDD操作，
  // 返回指定Key对应的元素形成的Seq。
  // 这个函数处理优化的部分在于，如果这个RDD包含分区器，
  // 则只会对应处理K所在的分区，然后返回由（K，V）形成的Seq。
  // 如果RDD不包含分区器，则需要对全RDD元素进行暴力扫描处理，
  // 搜索指定K对应的元素.
    def lookup(): Unit ={
      val cont = new SparkConf().setAppName("lookup").setMaster("local")
      val sc = new SparkContext(cont)
      sc.parallelize(Array((1,12),(2,11),(1,12),(2,11),(3,13))).lookup(1).foreach(println)
      sc.stop()
    }
  /*top（num：Int）（implicit ord：Ordering[T]）：Array[T]
自定义一个排序规则（倒序）
object MyOrdering extends Ordering[Int] {
def compare(a:Int, b:Int) = b compare a
}
top(10)(MyOrdering )
·top返回最大的k个元素。
·take返回前k个元素。
*/
    def top(): Unit ={
      val cont = new SparkConf().setAppName("top").setMaster("local")
      val sc = new SparkContext(cont)
      val list = List(1,2,3,4,5,6,7,8,9,10)
      sc.parallelize(list).top(3).foreach(println)
      sc.stop()
    }

  /**
    * old和reduce的原理相同，但是与reduce不同，相当于每个reduce时，迭代器取的第一个元素是zeroValue
    */
    def flod(): Unit ={
      val cont = new SparkConf().setAppName("flod").setMaster("local")
      val sc = new SparkContext(cont)
      val list = List(1,2,3)
      val i = sc.parallelize(list).fold(10)(_+_)
      println(i)
      sc.stop()
  }
  /*aggregate先对每个分区的所有元素进行aggregate操作，
  再对分区的结果进行fold操作。
　　aggreagate与fold和reduce的不同之处在于，
aggregate相当于采用归并的方式进行数据聚集，这种聚集是并行化的。
 而在fold和reduce函数的运算过程中，每个分区中需要进行串行处理，
 每个分区串行计算完结果，结果再按之前的方式进行聚集，并返回最终聚集结果。
*/
    def aggregate(): Unit ={
      def sqlOp(tuple1: (Int, Int), num: Int): (Int, Int) = {
        (tuple1._1+num,tuple1._2+1)
      }
      def combOp(tuple1: (Int,Int),tuple2: (Int,Int)): (Int,Int) = {
        (tuple1._1+tuple2._1,tuple1._2+tuple2._2);
      }
      val cont = new SparkConf().setAppName("aggregate").setMaster("local")
      val sc = new SparkContext(cont)
      val list = List(1,2,3,4,5,6,7,8,9,10)
      val tuple = sc.parallelize(list).aggregate((0,0))(sqlOp,combOp)
      println("平均值是"+ (tuple._1 / tuple._2))
      sc.stop()
    }

  def main(args: Array[String]): Unit = {
//    reduce()
//    collect()
//    count()
//    take()
//    saveAsTextFile()
//    countByKey()
//    collectAsMap()
//    lookup()
    top()
//    flod()
//    aggregate()
  }
}
